package com.atguigu.day04;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @description :
 * @autor :lyh
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.username)
                .process(new KeyedProcessFunction<String, ClickEvent, String>() {
                    private MapState<String, Long> urlCount;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        urlCount = getRuntimeContext().getMapState(
                                new MapStateDescriptor<String, Long>(
                                        "url-count",
                                        Types.STRING(),
                                        Types.LONG()
                                )
                        );
                    }

                    @Override
                    public void processElement(ClickEvent in, KeyedProcessFunction<String, ClickEvent, String>.Context ctx, Collector<String> out) throws Exception {
                        // in: ClickEvent("Marry","./home",1L)
                        // 如果 Marry 第一次访问 ./home
                        if (!urlCount.contains(in.url)) {
                            urlCount.put(in.url, 1L);
                        }
                        // 如果不是第一次访问
                        else {
                            urlCount.put(in.url, urlCount.get(in.url) + 1L);
                        }

                        // 输出结果
                        // 输出的是 in.username 对应的字典状态变量
                        StringBuilder result = new StringBuilder();
                        result.append(ctx.getCurrentKey() + "{\n");
                        for (String url : urlCount.keys()) {
                            result.append(" '" + url + "' => " + urlCount.get(url) + "\n");
                        }
                        result.append("}");
                        out.collect(result.toString());

                    }
                })
                .print();

        env.execute();
    }
}
